package homework2.miningPopularSong.reducer;

import homework2.miningPopularSong.program.MiningPopularSong;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Set;
import java.util.Map.Entry;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class ReduceRound2 extends Reducer<Text, Text, Text, Text> {

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		super.cleanup(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		
		Integer userCount = 0;
		Set<String> userCheck = new HashSet<>();
		Map<String, Integer> songCount = new HashMap<>();

		for (Text mappedValue : values) {
			String data[] = mappedValue.toString().split("\t");
			
			/* count users - if the user has already been inserted, he doesn't inserted another time */
			if (userCheck.add(data[0])) {
				userCount++;
			}

			/* count the times that songs are played */
			Integer timesCount = new Integer(data[2]);
			if (songCount.containsKey(data[1])) {
				songCount.put(data[1], songCount.get(data[1]) + timesCount);
			}
			else {
				songCount.put(data[1], timesCount);
			}
		}
		
		Integer k = context.getConfiguration().getInt(MiningPopularSong.KSONGS, 3);
		
		/* first, songs are sorted in ascending order */
		Queue<Song> ksongs = new PriorityQueue<>(k, Song.SongComparator);

		Iterator<Entry<String, Integer>> iterator = songCount.entrySet().iterator();
		
		/* take the name of country without $$ */
		key = new Text(key.toString().substring(2));

		/* insert first song */
		Entry<String, Integer> firstSong = iterator.next();
		ksongs.add(new Song(firstSong.getKey(), firstSong.getValue()));

		/* for each song played in a country */
		while (iterator.hasNext()) {
			Entry<String, Integer> song = iterator.next();

			if (ksongs.size() < k) {
				ksongs.add(new Song(song.getKey(), song.getValue()));
			} 
			else if (song.getValue() > (ksongs.peek().getNumberPlayed())) {
				/* remove the k-last element */
				ksongs.remove();
				/* add song */
				ksongs.add(new Song(song.getKey(), song.getValue()));
			}
			else {
				/* discard */
			}

			iterator.remove();
		}

		/* put the songs in descending order */ 
		Queue<Song> descendingOrderKsongs = new PriorityQueue<>(k, Song.ReverseSongComparator);
		
		while (!ksongs.isEmpty()) {
			descendingOrderKsongs.add(ksongs.poll());
		}
		
		/* format final output */ 
		String result = "";
		result += "\nPopular songs in " + key + "\t (based on " + userCount + " users)\n";
		
		while (!descendingOrderKsongs.isEmpty()) {
			Song song = descendingOrderKsongs.poll();
			String[] out = song.getName().split("\\$");
			if (out.length == 2)
				result += "\tSong:\t" + out[1] + " by " + out[0] + ", " + song.getNumberPlayed() + " plays\n";
		}
				
		context.write(new Text(""), new Text(result));		
	}

	@Override
	public void run(Context arg0) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		super.run(arg0);
	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		super.setup(context);
	}

}
